package job.gfp.ranks_v2.algorithm;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashSet;
import java.util.Set;

import job.gfp.ranks_v2.RanksV2_Data;
import mapred.config.JobConfig;

import org.apache.hadoop.fs.Path;

import util.FeatureMap;
import util.UrlUtil;

public class RanksV2_TrustRank extends RanksV2_PageRank {

    private static Set<String> seeds = null;

    @Override
    public String getName() {
        return "pTrustRank";
    }

    @Override
    public String[] getFields() {
        return new String[] { "Rank", "Bias", "OutDegree" };
    }

    @Override
    public RanksV2_Data initialize(String source, FeatureMap targets,
            JobConfig config) {
        RanksV2_Data ret = super.initialize(source, targets, config);
        
        if (isSeed(source, config)) {
            ret.set("Bias", config.get("TrustRank/Bias", 1));
        } else {
            ret.set("Bias", 0);
        }

        return ret;
    }

    private boolean isSeed(String source, JobConfig config) {
        try {
            return this.seeds(config).contains(UrlUtil.getHost(source));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    private Set<String> seeds(JobConfig config) throws IOException {
        if (seeds == null) {
            try {
                seeds = new HashSet<String>();
                BufferedReader br = new BufferedReader(new InputStreamReader(config
                        .getFileSystem().open(
                                new Path(config.get("TrustRank/Seeds")))));
                for (String line = br.readLine(); line != null; line = br.readLine()) {
                    seeds.add(line.trim());
                }
                seeds.remove("");
                br.close();
            } catch (IOException ioe) {
                seeds = null;
                throw ioe;
            }
        }
        return seeds;
    }

}
